package com.smsc.headend.task.engine.handler.impl.metercontrol;

import cn.hutool.core.convert.Convert;
import cn.hutool.core.date.SystemClock;
import cn.hutool.core.map.MapUtil;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.smsc.headend.module.data.enums.OperationGroup;
import com.smsc.headend.module.data.enums.OperationName;
import com.smsc.headend.module.metercontrol.enums.ControlState;
import com.smsc.headend.module.protocol.DeviceData;
import com.smsc.headend.module.report.entity.LogConnectDisConnectHis;
import com.smsc.headend.module.report.entity.LogModeSwitchHisSub;
import com.smsc.headend.module.report.entity.LogScheduleTaskDevice;
import com.smsc.headend.module.task.consts.TaskKafkaTopic;
import com.smsc.headend.module.task.dto.AtomicTask;
import com.smsc.headend.module.task.dto.Task;
import com.smsc.headend.module.task.enums.SubTaskStatus;
import com.smsc.headend.module.task.enums.TaskType;
import com.smsc.headend.task.engine.dao.TaskDAO;
import com.smsc.headend.task.engine.exception.UDISTaskHandlerException;
import com.smsc.headend.task.engine.handler.TaskHandler;
import com.smsc.headend.task.engine.mapper.LogConnectDisConnectHisMapper;
import com.smsc.headend.task.engine.service.FeignAssetManagementService;
import com.smsc.headend.task.engine.service.task.LogScheduleTaskDeviceService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;

import java.nio.charset.Charset;
import java.util.List;
import java.util.Map;

/**
 * @author wxh
 * @date 2021/1/6
 */
@Slf4j
public class RemoteActionHandler implements TaskHandler {
    @Autowired
    TaskDAO taskDAO;

    @Autowired
    FeignAssetManagementService assetManagementService;

    @Autowired
    KafkaTemplate kafkaTemplate;

    @Autowired
    LogConnectDisConnectHisMapper logConnectDisConnectHisMapper;

    @Autowired
    LogScheduleTaskDeviceService logScheduleTaskDeviceService;

    @Override
    public int handle(Task task, AtomicTask atomicTask, DeviceData cosemData, String actionResult) throws UDISTaskHandlerException {
        log.info("{} action result {}", atomicTask.getAtomicTaskNo(), actionResult);
        if (task.getTaskType().equals(TaskType.SwitchMeterPayment.toString())) {
            Map<String, Object> parameterMap = getParameterMap(task, atomicTask);
            Long taskId = MapUtil.getLong(parameterMap, "taskId");
            assetManagementService.updateStepStatus(taskId, task.getMeterId(), atomicTask.getGroupId().longValue() == OperationGroup.Disconnect.getId() ? LogModeSwitchHisSub.Step.Disconnect : LogModeSwitchHisSub.Step.Connect,
                    SubTaskStatus.Success.getCode(), null, SystemClock.now() / 1000, null);
        } else {
            // 拉闸子步骤最后为action操作. 成功后返回数据到redis给web 硬合闸最后一步是设置controlMode 这里不插入数据到redis
            if (atomicTask.getGroupId().longValue() == OperationGroup.Disconnect.getId()) {
                taskDAO.addTaskData(task.getTaskNo(), atomicTask.getAtomicTaskNo(), "result", "Success", 600l, Boolean.TRUE);
            }
            // 硬合闸目前最后一步是直接合闸action操作 这里加上判断是最后一个任务 适配之后硬合闸流程还要做setControlMode操作
            if (atomicTask.getGroupId().longValue() == OperationGroup.ConnectArmed.getId() && isSubTaskGroupLastAtomicTask(task, atomicTask)) {
                taskDAO.addTaskData(task.getTaskNo(), atomicTask.getAtomicTaskNo(), "result", "Success", 600l, Boolean.TRUE);
            }
            sendControlStateToAdapter(task, atomicTask);
            updateLog(task.getMessageId(), task.getMeterId(), LogConnectDisConnectHis.Result.success.getCode(), "");
            if (task.getTaskType().equals(TaskType.MeterPlanCdc.toString())) {
                logScheduleTaskDeviceService.sendMessage(task, LogScheduleTaskDevice.StatusEnum.Success.getCode(), LogScheduleTaskDevice.StatusDescEnum.CollectedSuccess.getCode());
            }
        }
        return 1;
    }

    @Override
    public int handleFailure(Task task, AtomicTask atomicTask, String message) {
        log.info("error {}, ", message);
        if (atomicTask.getGroupId() == OperationGroup.ConnectArmed.getId().intValue()
                || atomicTask.getGroupId() == OperationGroup.Disconnect.getId().intValue()
                || atomicTask.getGroupId() == OperationGroup.CheckCdc.getId().intValue()) {
            if (task.getTaskType().equals(TaskType.SwitchMeterPayment.toString())) {
                Map<String, Object> parameterMap = getParameterMap(task, atomicTask);
                Long taskId = MapUtil.getLong(parameterMap, "taskId");
                assetManagementService.updateStepStatus(taskId, task.getMeterId(), atomicTask.getGroupId() == OperationGroup.ConnectArmed.getId().intValue() ? LogModeSwitchHisSub.Step.Connect : LogModeSwitchHisSub.Step.Disconnect,
                        SubTaskStatus.Failed.getCode(), null, SystemClock.now() / 1000, String.format("handle %s fail, %s", OperationName.forId(atomicTask.getOperationId()).getName(), message));
            } else {
                String key = "message";
                Map<String, Object> m = JSONUtil.toBean(task.getParamJson(), Map.class);
                key = m.get("seq") != null ? Convert.toStr(m.get("seq")) : key;
                String stepOperationName = OperationName.forId(atomicTask.getOperationId()).getName();
                if (task.getIsODR()) {
                    taskDAO.addTaskData(task.getTaskNo(), atomicTask.getAtomicTaskNo(), key, String.format("step %s failed, [%s]", stepOperationName, message), 3600l, false);
                }
                updateLog(task.getMessageId(), task.getMeterId(), LogConnectDisConnectHis.Result.failed.getCode(), message);
            }
        }

        if (task.getTaskType().equals(TaskType.MeterPlanCdc.toString())) {
            logScheduleTaskDeviceService.sendMessage(task, LogScheduleTaskDevice.StatusEnum.Failed.getCode(), LogScheduleTaskDevice.StatusDescEnum.DecodeDataFailed.getCode());
        }
        return -1;
    }

    private void sendControlStateToAdapter(Task task, AtomicTask atomicTask) {
        JSONObject obj = JSONUtil.createObj();
        if (atomicTask.getGroupId().longValue() == OperationGroup.Disconnect.getId()) {
            obj.put("meterId", task.getMeterId());
            obj.put("value", Convert.toStr(ControlState.Disconnected.getCode()));
        } else if (atomicTask.getGroupId().longValue() == OperationGroup.Connect.getId()) {
            obj.put("meterId", task.getMeterId());
            obj.put("value", Convert.toStr(ControlState.Connected.getCode()));
        }
        obj.put("tv", SystemClock.now() / 1000);
        log.info("send control state to adapter: {}", JSONUtil.toJsonStr(obj));
        kafkaTemplate.send(TaskKafkaTopic.CLIENT_CONTROL_STATE, JSONUtil.toJsonStr(obj).getBytes(Charset.forName("utf-8")));
    }

    public void updateLog(String messageId, Long meterId, Integer result, String reason) {
        List<LogConnectDisConnectHis> logConnectDisConnectHisList = logConnectDisConnectHisMapper.selectList(Wrappers.<LogConnectDisConnectHis>lambdaQuery()
                .eq(LogConnectDisConnectHis::getMessageId, messageId).eq(LogConnectDisConnectHis::getDeviceId, meterId).orderByDesc(LogConnectDisConnectHis::getTvUpdate));
        if (null != logConnectDisConnectHisList && !logConnectDisConnectHisList.isEmpty()) {
            LogConnectDisConnectHis logConnectDisConnectHis = logConnectDisConnectHisList.get(0);
            logConnectDisConnectHis.setResult(result);
            logConnectDisConnectHis.setReason(reason);
            logConnectDisConnectHisMapper.updateById(logConnectDisConnectHis);
        }
    }
}
